package com.dyh.quickdemozwei.thp_reactor;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import lombok.Getter;
import lombok.Setter;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.function.Consumer;

/**
 * @Author: yhdong
 * @Date: 2021/11/12 16:12
 */
@Slf4j
public class Reactor1112_1 {




    @Setter@Getter
    public static class Node<T>{
        String code;
        Consumer<T> consumer;
        Consumer<T> doneConsumer;
        ThreadPoolExecutor poolExecutor;
        int maxBatch;
        volatile byte[] ctrl;//要能看到

        LinkedList<T> tasksQueue = new LinkedList();

        public Node(String code, Consumer<T> consumer, ThreadPoolExecutor poolExecutor, int maxBatch) {
            this.code = code;
            this.consumer = consumer;
            this.poolExecutor = poolExecutor;
            this.maxBatch = maxBatch;
            this.ctrl = new byte[maxBatch];
        }
    }

    private Map<String,Node> nodeMap = new HashMap<>();

    public void regist(Node node){
        nodeMap.put(node.getCode(),node);
    }

    public void start() {
        SelectThread selectThread =  new SelectThread();
        selectThread.start();
    }

    public <T> void apply(String code, List<T> tasks,Consumer doneConsumer){
        applyKenel(code,tasks,doneConsumer);
    }

    public <T> void applyKenelAsync(String code, List<T> tasks,Consumer doneConsumer){
        Node node = nodeMap.get(code);
        if (node==null)
            return;

        node.setDoneConsumer(doneConsumer);
        node.getTasksQueue().addAll(tasks);
    }

    public <T> void applyKenel(String code, List<T> tasks,Consumer doneConsumer){
        Node node = nodeMap.get(code);
        if (node==null)
            return;

        f1:
        for (;;){
            if (checkAllInit(node.getCtrl())){
                if (tasks.size()<=node.getMaxBatch()){
                    node.setDoneConsumer(doneConsumer);
                    for (int i = 0;i<tasks.size();i++){
                        node.getCtrl()[i]=1;
                        T task = tasks.get(i);
                        node.getPoolExecutor().submit(()->node.getConsumer().accept(task));
                    }
                    break;
                }
                else {
                    List<List<T>> splitTasks = Lists.partition(tasks,node.getMaxBatch());
                    for (int i = 0;i<splitTasks.size();i++){
                        if (i==splitTasks.size()-1)
                            node.setDoneConsumer(doneConsumer);

                        List<T> splitTask = splitTasks.get(i);
                        for (;;){
                            if (checkAllInit(node.getCtrl())){
                                for (int j = 0;j<splitTask.size();j++){
                                    node.getCtrl()[j]=1;
                                    T task = splitTask.get(j);
                                    node.getPoolExecutor().submit(()->node.getConsumer().accept(task));
                                }
                                break;
                            }
                        }

                        if (i==splitTasks.size()-1)
                            break f1;
                    }
                }
            }

        }
    }


        public class SelectThread extends Thread{
        @SneakyThrows
        @Override
        public void run() {
            for (;;){
                for (Map.Entry<String, Node> entry : nodeMap.entrySet()) {
                    Node node = entry.getValue();
                    if ( checkAllRuned(node.getCtrl())){
                        log.info("{} selected",Thread.currentThread().getName());
                        if (node.getDoneConsumer()!=null){
                            node.getDoneConsumer().accept(null);
                            node.setDoneConsumer(null);
                        }
                        resetInit(node.getCtrl());
                    }
                }
                Thread.sleep(500L);
            }
        }
    }



    boolean checkAllRuned( byte[] ctrl){
        for (byte b : ctrl) {
            if (b==0)
                return false;
        }
        return true;
    }

    boolean checkAllInit( byte[] ctrl){
        for (byte b : ctrl) {
            if (b==1)
                return false;
        }
        return true;
    }

    void resetInit(byte[] ctrl){
        for (int i = 0; i<ctrl.length;i++){
            ctrl[i]=0;
        }

    }


}
